package db.app

import com.cw.realtime.common.base.FlinkRunner.{RunnerConfig, SqlRunnerConfig, createTopicDbTable, runSql}
import com.cw.realtime.common.constant.Constant._
import com.cw.realtime.common.util.SqlUtil
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment

import java.time.Duration

object DwdTradeOrderDetail {


  def main(args: Array[String]): Unit = {

    implicit val config = SqlRunnerConfig(TOPIC_DWD_TRADE_ORDER_DETAIL, 10024)

    runSql { (_, tableEnv) =>

      tableEnv.getConfig.setIdleStateRetention(Duration.ofSeconds(3L))

      createTopicDbTable(tableEnv)

      val orderDetail = filterOrderDetail(tableEnv)
      tableEnv.createTemporaryView("order_detail", orderDetail)


      val orderInfo = filterOrderInfo(tableEnv)
      tableEnv.createTemporaryView("order_info", orderInfo)


      val oda = filterOrderDetailActivity(tableEnv)
      tableEnv.createTemporaryView("order_detail_activity", oda)

      val odc = filterOrderDetailCoupon(tableEnv)
      tableEnv.createTemporaryView("order_detail_coupon", odc)

      val joined = joinTable(tableEnv)

//      joined.execute().print()

      createKafkaSink(tableEnv)

      joined.insertInto(TOPIC_DWD_TRADE_ORDER_DETAIL).execute()


    }


  }

  def createKafkaSink(tableEnvironment: StreamTableEnvironment) = {
    tableEnvironment.executeSql(
      s"""
create table $TOPIC_DWD_TRADE_ORDER_DETAIL (
  id                string,
  user_id           string,
  province_id       string,
  order_id          string,
  sku_id            string,
  activity_id       string,
  activity_rule_id  string,
  coupon_id         string,
  sku_name          string,
  order_price       string,
  sku_num           string,
  create_time       string,
  split_total_amount        string,
  split_activity_amount     string,
  split_coupon_amount       string,
  ts                bigint,
  PRIMARY KEY (id) NOT ENFORCED

) ${SqlUtil.getUpsertKafkaSinkSql(TOPIC_DWD_TRADE_ORDER_DETAIL)}

      """)
  }


  def joinTable(tableEnvironment: StreamTableEnvironment) = {
    tableEnvironment.sqlQuery(
      """
select
  od.id,
  user_id,
  province_id,
  order_id,
  sku_id,
  activity_id,
  activity_rule_id,
  coupon_id,
  sku_name,
  order_price,
  sku_num,
  create_time,
  split_total_amount,
  split_activity_amount,
  split_coupon_amount,
  od.ts ts

from order_detail od
join order_info oi on od.order_id = oi.id
left join order_detail_activity oda on od.id = oda.id
left join order_detail_coupon odc on od.id = odc.id
        """
    )

  }


  def filterOrderDetail(tableEnv: StreamTableEnvironment) = {
    tableEnv.sqlQuery(
      """
select
  `data`['id']                     id,
  `data`['order_id']               order_id,
  `data`['sku_id']                 sku_id,
  `data`['sku_name']               sku_name,
  `data`['order_price']            order_price,
  `data`['sku_num']                sku_num,
  `data`['create_time']            create_time,
  `data`['split_total_amount']     split_total_amount,
  `data`['split_activity_amount']  split_activity_amount,
  `data`['split_coupon_amount']    split_coupon_amount,
  ts

from topic_db
where
`database` = 'gmall' and
`table` = 'order_detail' and
 `type` = 'insert'
        """)
  }


  def filterOrderInfo(tableEnv: StreamTableEnvironment) = {
    tableEnv.sqlQuery(
      """
select
  `data`['id'] id,
  `data`['user_id'] user_id,
  `data`['province_id'] province_id,
  ts

from topic_db
where
`database` = 'gmall' and
`table` = 'order_info' and
 `type` = 'insert'

        """)
  }


  def filterOrderDetailActivity(tableEnv: StreamTableEnvironment) = {

    tableEnv.sqlQuery(
      """
select
  `data`['order_detail_id']   id,
  `data`['activity_id']        activity_id,
  `data`['activity_rule_id']   activity_rule_id
from topic_db
where
`database` = 'gmall' and
`table` = 'order_detail_activity' and
 `type` = 'insert'
      """
    )
  }


  def filterOrderDetailCoupon(tableEnvironment: StreamTableEnvironment) = {
    tableEnvironment.sqlQuery(
      """
select
  `data`['order_detail_id']   id,
  `data`['coupon_id']   coupon_id
from topic_db
where
  `database` = 'gmall' and
  `table` = 'order_detail_coupon' and
   `type` = 'insert'
    """
    )
  }
}
